package com.spi.service;

import com.spi.config.MyWebSocketHandler;
import com.spi.init.ServerInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.File;
import java.io.FileInputStream;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

@Service
@Slf4j
public class AsyncService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate ;

    @Lazy //懒加载 防止循环依赖报错
    @Autowired
    private AsyncService asyncService ;

    @Autowired
    private RedissonClient redissonClient ;

    @Autowired
    private ServerInfo serverInfo ;

    @Autowired
    private RedisTemplate redisTemplate ;

    @Value("${rocketmq.consumer.group4.tag}")
    private String ymlAddress ;

    @Async
    public void doLongTask(String sessionId) {
        try {
            // 模拟耗时操作
            Thread.sleep(5000);
            MyWebSocketHandler.sendMessageToClient(sessionId, "后台处理完成111");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public boolean removeSessionData(String socketid) {
        if (!StringUtils.isEmpty(socketid)){
            //移除容器内元素
            if (MyWebSocketHandler.sessions.containsKey(socketid)){
                MyWebSocketHandler.sessions.remove(socketid);
                System.out.println("移除容器内元素。。");
                log.info("移除容器内元素。。");
                return true ;
            }
        }
        return false ;
    }

    public void doAsyncTask(String sockId,String contendId){
        RLock lock = null ;
        String lockName = "survey:autoBuildCon:" + contendId ;
        try {
            lock = redissonClient.getLock(lockName);
            if (lock.tryLock()) {
                Thread.sleep(5000);
                //1: 进行加锁动作 redission分布式锁

                //2：获取影像信息列表 调用ocr接口 异步多线程调用
                String ocrMsg = getOCRMsg(contendId);
                //3: 将ocr识别文字汇总 去调用deepseekr1推理模型

                //4：任务执行完毕 获取到调查结论 进行落库
                //检查当前线程是否仍然持有锁
                if (!lock.isHeldByCurrentThread()){
                    //锁已被释放 或锁不被当前线程锁持有 ，任务终止 不落库 也不去通知了
                    return;
                }else {
                    //执行落库逻辑

                }
                //5：通知websocket任务执行完毕 需用到mq广播模式 仅本地持有sockid的节点才能进行通知
                notifyWebSocket(sockId,ocrMsg);
                //6：释放锁
            }else {
                System.out.println("获取锁失败。。。");
                log.info("获取锁失败。。。");
            }
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("doAsyncTask方法异常：" + e.getMessage());
            log.info("doAsyncTask方法异常：" + e.getMessage());
        }finally {
            //以线程为维度释放锁，防止其它未获取到锁的线程 将原有获取锁正在执行任务的线程的锁给释放了
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }

    }

    //获取影像信息列表并调用ocr接口
    private String getOCRMsg(String contendId){
        List<String> base64List = new ArrayList<>() ;
        StringBuffer stringBuffer = new StringBuffer() ;
        String corStringList = "" ;
        try {
            File file = new File("D:\\photo");
            if (file.exists() && file.isDirectory()){
                File[] files = file.listFiles();
                if (files != null){
                    for (File file1 : files) {
                        File file2 = new File(file1.getAbsolutePath()) ;
                        FileInputStream fileInputStream = new FileInputStream(file2) ;
                        byte[] imageData = new byte[(int) file2.length()];
                        fileInputStream.read(imageData);
                        base64List.add(Base64.getEncoder().encodeToString(imageData));
                    }
                }
            }
            //开启异步线程调用ocr接口
            List<CompletableFuture<String>> collect = base64List.stream()
                    .map(asyncService::callAPI)
                    .collect(Collectors.toList());
            CompletableFuture<Void> allFuture = CompletableFuture.allOf(collect.toArray(new CompletableFuture[0]));
            CompletableFuture<String> stringCompletableFuture = allFuture
                    .thenApply(v -> collect.stream()
                            .map(s -> {
                try {
                    return s.get();
                } catch (Exception e) {
                    e.printStackTrace();
                    return null;
                }
            })
                            .filter(result -> result != null)
                            .collect(Collectors.joining()));
            //获取ocr远程调用后拼接的字符串
            corStringList = stringCompletableFuture.get();
            System.out.println("拼接字符串如下：" + corStringList);

        }catch (Exception e){
            e.printStackTrace();
            System.out.println("getOCRMsg方法执行出现异常。。" + e.getMessage());
        }

        return  corStringList ;
    }

    @Async("JobServiceExecutor")
    public CompletableFuture<String> callAPI(String base64Image){
        try {
            Thread.sleep(3000);
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("callAPI模拟ocr调用操作报错==" + e.getMessage());
        }
        return CompletableFuture.completedFuture(base64Image.substring(0,10)) ;
    }

    //通知websocket任务执行完毕
    private void notifyWebSocket(String sockId,String message){
        //发送mq消息
        System.out.println("notifyWebSocket方法执行即将发送广播模式消息" + sockId);
        log.info("notifyWebSocket方法执行即将发送广播模式消息" + sockId);
        Map<String,String> map = new HashMap<>() ;
        map.put("sockId",sockId);
        map.put("message","r1自动化生成调查结论为：" + message);
        Message<Map<String, String>> mapMessage = MessageBuilder.withPayload(map).build();
        String ipAddr  = String.valueOf(redisTemplate.opsForValue().get(sockId));
        System.out.println("mq准备通知websocket发送消息对应标签tag为：" + ipAddr);
        log.info("mq准备通知websocket发送消息对应标签tag为：" + ipAddr);
        System.out.println("yml对应tag地址：" + ymlAddress);
        log.info("yml对应tag地址：" + ymlAddress);
        rocketMQTemplate.asyncSend("survey-topic-notifyWebSocket:" + ipAddr, mapMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("survey-topic-notifyWebSocket==mq生成消息id为"+sendResult.getMsgId());
                log.info("survey-topic-notifyWebSocket==mq生成消息id为"+sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("survey-topic-notifyWebSocket投递消息失败。。。");
                log.info("survey-topic-notifyWebSocket投递消息失败。。。");
            }
        });
    }


}
